package com.bw.wjw.job;

import com.bw.wjw.util.AnalyzerUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import java.util.List;

public class JtpLogSearchKeywordMinuteWindowDwsJob {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv() ;

        // 2. 输入表-input：映射到Kafka消息队列
        createInputTable(tabEnv);
        // tabEnv.executeSql("SELECT * FROM dwd_log_page_view_log_kafka_source LIMIT 10").print();

        // 3. 数据处理-select
        Table reportTable = handle(tabEnv);

        // 4. 输出表-output：映射到Clickhouse表
        createOutputTable(tabEnv) ;

        // 5. 保存数据-insert
        saveToClickHouse(tabEnv, reportTable) ;
    }

    private static Table handle(TableEnvironment tabEnv){
        // s1-获取搜索词和搜索时间
        Table searchLogTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    page['item'] AS full_word\n" +
                        "    , row_time\n" +
                        "FROM dwd_log_page_view_log_kafka_source\n" +
                        "WHERE page['item'] IS NOT NULL\n" +
                        "    AND page['last_page_id'] = 'search'\n" +
                        "    AND page['item_type'] = 'keyword'"
        );
//         searchLogTable.execute().print();
        tabEnv.createTemporaryView("search_log_table", searchLogTable);

        // s2-使用自定义UDTF函数，对搜索词进行中文分词
        tabEnv.createTemporarySystemFunction("ik_analyzer_udtf", IkAnalyzerFunction.class);
        Table wordLogTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    full_word\n" +
                        "    , keyword\n" +
                        "    , row_time\n" +
                        "FROM search_log_table,\n" +
                        "   LATERAL TABLE(ik_analyzer_udtf(full_word)) AS T(keyword)"
        );
//        wordLogTable.execute().print();
        tabEnv.createTemporaryView("word_log_table", wordLogTable);

        // s3-设置窗口进行分组、聚合计算
        Table reportTable = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    TUMBLE_START(row_time, INTERVAL '1' MINUTES) AS window_start_time\n" +
                        "    , TUMBLE_END(row_time, INTERVAL '1' MINUTES) AS window_end_time\n" +
                        "    , keyword\n" +
                        "    , count(keyword) AS keyword_count\n" +
                        "    , UNIX_TIMESTAMP() * 1000 AS ts\n" +
                        "FROM word_log_table\n" +
                        "GROUP BY\n" +
                        "    TUMBLE(row_time, INTERVAL '1' MINUTES ),\n" +
                        "    keyword"
        );
//         reportTable.execute().print();

        // 返回计算结果
        return reportTable;
    }


    private static void createOutputTable(TableEnvironment tabEnv){
        tabEnv.executeSql(
                "CREATE TABLE dws_log_search_keyword_window_report_clickhouse_sink (" +
                        "    `window_start_time` STRING COMMENT '窗口开始日期时间'," +
                        "       `window_end_time` STRING COMMENT '窗口结束日期时间'," +
                        "    `keyword` STRING COMMENT '搜索关键词'," +
                        "    `keyword_count` BIGINT COMMENT '搜索关键词被搜索次数'," +
                        "    `ts` BIGINT COMMENT '数据产生时间戳'" +
                        ") WITH (" +
                        "    'connector' = 'clickhouse'," +
                        "    'url' = 'jdbc:clickhouse://node103:8123/jtp_log_report'," +
                        "    'table' = 'dws_log_search_keyword_window_report'," +
                        "    'username' = 'default'," +
                        "    'password' = ''," +
                        "    'format' = 'json'" +
                        ")"
        );
    }


    private static void saveToClickHouse(TableEnvironment tabEnv, Table table){
        // 1将Table注册为表
        tabEnv.createTemporaryView("report_table", table);
        // 2插入数据
        tabEnv.executeSql(
                "INSERT INTO dws_log_search_keyword_window_report_clickhouse_sink\n" +
                        "SELECT\n" +
                        "    DATE_FORMAT(window_start_time, 'yyyy-MM-dd HH:mm:ss') AS window_start_time\n" +
                        "    , DATE_FORMAT(window_end_time, 'yyyy-MM-dd HH:mm:ss') AS window_end_time\n" +
                        "    , keyword\n" +
                        "    , keyword_count\n" +
                        "    , ts\n" +
                        "FROM report_table"
        );
    }


    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE dwd_log_page_view_log_kafka_source (" +
                        "    `common` MAP<STRING, STRING> COMMENT '公共环境信息'," +
                        "    `page` MAP<STRING, STRING> COMMENT '页面信息'," +
                        "    `ts` BIGINT," +
                        "    row_time AS TO_TIMESTAMP(FROM_UNIXTIME(ts / 1000, 'yyyy-MM-dd HH:mm:ss.SSS'))," +
                        "    WATERMARK FOR row_time AS row_time - INTERVAL '0' MINUTE" +
                        ") WITH (" +
                        "    'connector' = 'kafka'," +
                        "    'topic' = 'dwd-log-page-view-log'," +
                        "    'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092'," +
                        "    'properties.group.id' = 'gid_dws_log_search_keyword'," +
                        "    'scan.startup.mode' = 'earliest-offset'," +
                        "    'format' = 'json'," +
                        "    'json.fail-on-missing-field' = 'false'," +
                        "    'json.ignore-parse-errors' = 'true'" +
                        ")"
        );
    }


    private static TableEnvironment getTableEnv(){
        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings) ;
        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // 3返回对象
        return tabEnv;
    }

    @FunctionHint(output = @DataTypeHint("ROW<keyword STRING>"))
    public static class IkAnalyzerFunction extends TableFunction<Row> {
        public void eval(String fullWord) throws Exception {
            // 中文分词
            List<String> list = AnalyzerUtil.ikAnalyzer(fullWord);
            // 循环遍历输出
            for (String keyword : list) {
                collect(Row.of(keyword));
            }
        }
    }




}
